package com.zc.component.network.tcp.client;

import com.zc.component.network.core.DefaultNetworkType;
import com.zc.component.network.core.NetworkType;
import com.zc.component.network.tcp.parser.PayloadParser;
import com.zc.component.network.tcp.server.TcpServer;
import io.vertx.core.net.NetClient;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.EmitterProcessor;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;

import java.util.function.Function;

@Slf4j
public class VertxTcpClientServer implements TcpServer {

    @Getter
    private final String id;
    private final EmitterProcessor<TcpClient> processor = EmitterProcessor.create(false);
    private final FluxSink<TcpClient> sink = processor.sink(FluxSink.OverflowStrategy.BUFFER);
    public volatile NetClient client;
    volatile PayloadParser payloadParser;

    public VertxTcpClientServer(String id) {
        this.id = id;
    }

    @Override
    public boolean isAlive() {
        return client != null;
    }

    @Override
    public boolean isAutoReload() {
        return false;
    }

    private void execute(Runnable runnable) {
        try {
            runnable.run();
        } catch (Exception e) {
            log.warn("close tcp client error", e);
        }
    }

    @Override
    public NetworkType getType() {
        return DefaultNetworkType.TCP_CLIENT;
    }

    /**
     * 订阅客户端连接
     *
     * @return 客户端流
     * @see TcpClient
     */
    @Override
    public Flux<TcpClient> handleConnection() {
        return processor
                .map(Function.identity());
    }

    @Override
    public void shutdown() {
        log.debug("tcp client [{}] disconnect", getId());
        synchronized (this) {
            if (null != client) {
                execute(client::close);
                client = null;
            }
            if (null != payloadParser) {
                execute(payloadParser::close);
                payloadParser = null;
            }
        }
    }

    public void setClient(NetClient client) {
        if (this.client != null && this.client != client) {
            this.client.close();
        }
        this.client = client;
    }

    /**
     * 设置客户端消息解析器
     *
     * @param payloadParser 消息解析器
     */
    public void setRecordParser(PayloadParser payloadParser) {
        if (null != this.payloadParser && this.payloadParser != payloadParser) {
            this.payloadParser.close();
        }
        this.payloadParser = payloadParser;
    }

    public void sinkConnectClient(TcpClient client) {
        sink.next(client);
    }
}
